package com.atguigu.gmall.realtime.app.dwd;

import com.atguigu.gmall.realtime.app.BaseSQLApp;
import com.atguigu.gmall.realtime.common.Constant;
import com.atguigu.gmall.realtime.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @Author lzc
 * @Date 2022/12/6 08:54
 */
public class Dwd_02_DwdTradeCartAdd extends BaseSQLApp {
    public static void main(String[] args) {
        new Dwd_02_DwdTradeCartAdd().init(
            3002,
            2,
            "Dwd_02_DwdTradeCartAdd"
        );
        
    }
    
    @Override
    protected void handle(StreamExecutionEnvironment env,
                          StreamTableEnvironment tEnv) {
        
        // 1. 通过 ddl 去读取 ods_db 数据
        readOdsDb(tEnv, "Dwd_02_DwdTradeCartAdd");  // 后面几乎每个 app 都会用到这个表
        // 2. 读取字段
        readBaseDic(tEnv);
        
        // 2. 过滤出 cart_info 表数据
        Table cartInfo = tEnv.sqlQuery("select " +
                                           "data['id'] id," +
                                           "data['user_id'] user_id," +
                                           "data['sku_id'] sku_id," +
                                           "data['source_id'] source_id," +
                                           "data['source_type'] source_type," +
                                           "if(`type` = 'insert'," +
                                           "data['sku_num'],cast((cast(data['sku_num'] as int) - cast(`old`['sku_num'] as int)) as string)) sku_num," +
                                           "ts, " +
                                           "pt " +
                                           "from ods_db " +
                                           "where `database`='gmall2022' " +
                                           "and `table`='cart_info' " +
                                           "and (" +
                                           "   `type`='insert' or (" +
                                           "        `type`='update' " +
                                           "          and `old`['sku_num'] is not null " +
                                           "          and cast(`data`['sku_num'] as int)>cast(`old`['sku_num'] as int) " +
                                           "    )" +
                                           ")");
        tEnv.createTemporaryView("cart_info", cartInfo);
    
        Table result = tEnv.sqlQuery("select " +
                                        "ci.id," +
                                        "user_id," +
                                        "sku_id," +
                                        "source_id," +
                                        "source_type," +
                                        "dic_name source_type_name," +
                                        "sku_num," +
                                        "ts " +
                                        "from cart_info ci " +
                                        "join base_dic for system_time as of ci.pt as dic " +
                                        "on ci.source_type=dic.dic_code ");
        
        // 写出到 dwd 层
        tEnv.executeSql("create table dwd_trade_cart_add(" +
                            "id string," +
                            "user_id string," +
                            "sku_id string," +
                            "source_id string," +
                            "source_type_code string," +
                            "source_type_name string," +
                            "sku_num string," +
                            "ts bigint" +
                            ")" + SQLUtil.getKafkaSinkDDL(Constant.TOPIC_DWD_TRADE_CART_ADD, "json"));
    
        result.executeInsert("dwd_trade_cart_add");
    
    }
    
    
}
/*
------
select sum(sku_num) from  group by sku_id
sku_num 这个字段,怎么取?
如果是新增: 直接取值
如果更新:
sku_id     sku_num
10            3
10            5 而是 5-3=2

表的 join


------------------
加购事务实时表
使用流的方式
	业务比较复杂, 数据格式也比较多样

使用 sql
	结构化的数据, 业务不是很复杂

	在企业中用的比较多. 简单


1. 消费 kakfa 的业务数据: ods_db
	使用 sql: ddl

2. 过滤数据( sql 查询)

3. 维度退化(维度冗余)
	sql join: 3个课时

	cart_info 和 base_dic

3. 加购详情表数据写入到 kakfa 中(dwd 层)
 */